package yredis

import (
	"context"
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/utils"
	"dgo/plg/plgnet"
	"fmt"
	"gitee.com/ymofen/supervalue"
	"strings"
	"sync/atomic"
	"time"
)

/*
*

	订阅数据, 进行输出, 只适用一个连接进行订阅
	   一般进行匹配订阅
*/
type RedisPubPlg struct {
	send_err_n  uint32
	send_n      uint32
	push_fail_n uint32

	last_pub_t int64
	PlgBase

	msglst *utils.SyncCycleList

	workers *utils.MultiGroupTask
	pubsync bool

	sublst string

	redisInner *RedisInner
	redis_ctx  context.Context

	monitor_send *utils.MonitorSpeedSizeRec
	monitor_recv *utils.MonitorSpeedSizeRec

	pub_topic_lst utils.GSyncMapGroup

	last_pub_warn      string
	last_recv          string
	last_recv_t        time.Time
	last_recv_workflag byte
	last_ping_msg      string
}

/*
{
"id": "xxx",
"type": "yredis.pub",
"conf": {
"addr":"119.96.169.117:6379",
"pub-sync":true
}
}
*/
func (this *RedisPubPlg) ReloadConf(conf *wrapper.SuperValue) error {
	if conf == nil {
		return nil
	}

	if this.redisInner == nil {
		this.redisInner = NewRedisInner()
	}

	if this.redis_ctx == nil {
		this.redis_ctx = context.Background()
	}

	if this.workers == nil {
		this.workers = utils.NewMultiGroupTask()
		this.workers.ConfigWorkType(2)
		this.workers.ConfigMinWork(1)
		this.workers.ConfigMaxWork(56)
		this.workers.Start()
	}

	this.pubsync = conf.BoolByName("pub-sync", true)

	if this.monitor_send == nil {
		this.monitor_send = utils.NewMonitorSpeedSizeRec()
		this.monitor_recv = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitorRec(this.monitor_send, this.monitor_send)
		utils.Monitor.AddMonitorRec(this.monitor_recv, this.monitor_recv)
	}

	if this.redisInner.ReloadConf(conf) {
		this.redisInner.RequestDis("配置改变")
		this.redisInner.CheckNewClt()
		time.AfterFunc(time.Second, func() {
			err := this.redisInner.Ping()
			if err != nil {
				this.last_ping_msg = fmt.Sprintf("[%s]Reids(%s) PING err: %s", utils.NowString(), this.redisInner.RedisAddr(), err.Error())
				this.LogWarnMessage("与Redis建立连接失败, PING err: %s", err.Error())
			} else {
				this.last_ping_msg = fmt.Sprintf("[%s]Reids(%s) PING succ", utils.NowString(), this.redisInner.RedisAddr())
				this.LogInfoMessage("与Redis建立连接成功")
			}
		})
	}

	if this.msglst == nil {
		this.msglst = utils.NewSyncCycleList()
	}

	utils.DefaultPatrolTask().AddTask(this, time.Second*30, func(id interface{}, args ...interface{}) {
		//utils.DefaultWorkers().PostTaskFunc(this, this.innerWork)
	})

	return nil
}

func (this *RedisPubPlg) innerPub(topic string, data interface{}) error {
	clt := this.redisInner.clt
	if clt == nil {
		atomic.AddUint32(&this.send_err_n, 1)
		return fmt.Errorf("尚未建立连接")
	}
	if data != nil {
		if buf, ok := utils.TryGetBuf(data); ok {
			_, err := clt.Publish(this.redis_ctx, topic, buf).Result()
			if err != nil {
				this.last_pub_warn = fmt.Sprintf("[%s]topic:%s, err:%s", utils.NowString(), topic, err.Error())
				atomic.AddUint32(&this.send_err_n, 1)
				return err
			}
			this.last_pub_t = time.Now().Unix()
			this.monitor_send.DeltaBuf(buf)
			return nil
		} else {
			atomic.AddUint32(&this.send_err_n, 1)
			return fmt.Errorf("未知的数据")
		}
	} else {
		atomic.AddUint32(&this.send_err_n, 1)
		return fmt.Errorf("数据为nil")
	}
}

func (this *RedisPubPlg) Pub(topic string, max int, args ...interface{}) int {
	if len(args) < 0 {
		return -1
	}
	data := args[0]
	if this.pubsync {
		err := this.innerPub(topic, data)
		utils.TryCallBack(data, this, err)
		if err != nil {
			return -1
		}
		return 1
	} else {
		err := this.workers.PostTaskFunc(topic, func() {
			errinner := this.innerPub(topic, data)
			utils.TryCallBack(data, this, errinner)
		})
		if err != nil {
			utils.TryCallBack(data, this, err)
			return -1
		}
		return 1
	}
}

func (this *RedisPubPlg) DispatchData(channelid string, sender interface{}, data interface{}, tag interface{}) (err error) {
	if this.pubsync {
		err = this.innerPub(channelid, data)
		utils.TryCallBack(data, this, err)
		return
	} else {
		err = this.workers.PostTaskFunc(channelid, func() {
			errinner := this.innerPub(channelid, data)
			utils.TryCallBack(data, this, errinner)
		})
		if err != nil {
			utils.TryCallBack(data, this, err)
		}
		return
	}

}

func (this *RedisPubPlg) Close() (err error) {
	if this.redisInner != nil {
		this.redisInner.Close()
	}
	this.PlgBase.Close()
	utils.Monitor.RemoveMonitorRec(this.monitor_recv)
	utils.Monitor.RemoveMonitorRec(this.monitor_send)
	utils.DefaultPatrolTask().DelTask(this)
	this.redisInner = nil
	return
}

func (this *RedisPubPlg) OnHttpStatus(rec *plgnet.PlgHttpRec) (rval interface{}, errcode int, err error) {
	itype := rec.QryAsInt("type", 0)
	if itype == 99 {
		json := wrapper.NewSVObject()
		clt := this.redisInner.clt
		//sub := this.redisclt_sub
		if clt != nil {
			json.SetKeyValue("redis-addr", clt.Options().Addr)
			json.SetKeyValue("redis-pool-status", fmt.Sprintf("idle/total :%d/%d", clt.PoolStats().IdleConns, clt.PoolStats().TotalConns))
		}
		json.SetKeyValue("last-disp-warn", this.last_pub_warn)
		json.SetKeyValue("last-ping", this.last_ping_msg)
		json.SetKeyValue("monitor-send", this.monitor_send.Info())
		json.SetKeyValue("pub-sync", this.pubsync)
		json.SetKeyValue("workers", this.workers.StatusSimpleString())

		json.SetKeyValues("now", utils.NowString())
		if rec.QryAsInt("json", 0) == 1 {
			PrjHttpUtils.ResponseAsJSON(rec.RespW, json.AsJSONStringEx(false, true))
		} else {
			PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.AsJSONString(false))
		}
	} else if itype == 98 {
		rec.ResponseSyncListNEx(this.msglst)
	} else if itype == 2 {
		json := wrapper.NewSVArray()
		secs := time.Duration(rec.QryAsInt("secs", 0)) * time.Second

		prefix := rec.QryAsString("prefix", "")
		trimprefix := rec.QryAsString("trimprefix", "")

		suffix := rec.QryAsString("suffix", "")
		trimsuffix := rec.QryAsString("trimsuffix", "")

		this.pub_topic_lst.Range(func(key, value interface{}) bool {
			t0 := value.(time.Time)
			if time.Since(t0) < secs {
				keystr := key.(string)
				if len(prefix) > 0 {
					if !strings.HasPrefix(keystr, prefix) {
						return true
					}
				}
				if len(trimprefix) > 0 {
					keystr = strings.TrimPrefix(keystr, trimprefix)
				}

				if len(suffix) > 0 {
					if !strings.HasSuffix(keystr, suffix) {
						return true
					}
				}
				if len(trimsuffix) > 0 {
					keystr = strings.TrimSuffix(keystr, trimsuffix)
				}

				itm := json.AppendObject()
				itm.SetKeyValue("id", keystr)
				itm.SetKeyValue("last_recv_t", utils.DateTimeString2(t0))
			}
			return true
		})
		if rec.QryAsInt("json", 0) == 1 {
			PrjHttpUtils.ResponseAsJSON(rec.RespW, json.AsJSONStringEx(false, false))
		} else {
			PrjHttpUtils.JSON2Table(rec.RespW, "", json.AsJSONString(false))
		}
	} else {
		var sb utils.BytesBuilder
		PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")
		PrjHttpUtils.AppendIndexBody(&sb, "msglist", "status?type=98&n=100&searchval=")
		PrjHttpUtils.ResponseBodyHtml(rec.RespW, sb.String())
	}
	return
}

func (this *RedisPubPlg) LogInfoMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_INFO, 1, s, args...)
}

func (this *RedisPubPlg) LogWarnMessage(s string, args ...interface{}) {
	this.msglst.LogMsg(s, args...)
	utils.Log.Msgf(this.PlugId, utils.LogLV_WARN, 1, s, args...)
}

func CreateRedisPubPlgFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &RedisPubPlg{}

	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.PlugId = str
		}
	}
	rval = obj
	return
}
